package app.subject

import com.hankcs.hanlp.HanLP
import com.hankcs.hanlp.seg.common.{Term => HanLPTerm}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.JavaConversions._


/**
  * Created by duanxiping on 2016/11/1.
  */
object SubjectExtract {

  //spark,hive
  var hiveContext: HiveContext = null
  var sc: SparkContext = null
  var sqlContext: SQLContext = null

  /*
  project.mdl_weblog_owl_tokenized_post_hi
  project.mdl_weblog_owl_tokenized_thread_hi
  */
  val src_table = "project.mdl_weblog_owl_tokenized_thread_hi"
  val dst_table = "project.dxp_mdl_weblog_owl_tokenized_thread_extract"

  def getDataFromHive(table: String, stat_date: BigInt, stat_hour: BigInt): DataFrame = {
    val select_sql = s"select topicid,topicname from $table where stat_date=$stat_date " +
      "and title is not null and length(title)>2"
    hiveContext.sql(select_sql)
  }

  def initSpark(appname: String): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName(appname)
    sc = new SparkContext(sparkConf)
    hiveContext = new HiveContext(sc)
    sqlContext = new SQLContext(sc)
  }

  def main(args: Array[String]): Unit = {
    initSpark("Subject Extract")
    val stat_date: BigInt = args(0).toInt
    val stat_hour: BigInt = args(1).toInt

    val data_df = getDataFromHive(src_table, stat_date, stat_hour)


  }

  def calWordScore(df:DataFrame):Array[(String,(Int,Int))] = {
    /*词性网址
    http://www.hankcs.com/nlp/part-of-speech-tagging.html
     */
    //停用pos
    val filterPos = Array("c","f","uj","p","w")
    df.flatMap(r=>{
      val id = r.get(0).toString
      val sen = r.get(1).toString

    })
  }
  def segMsgWithNature(msg: String): Array[(String,String)] = {
    val wordarr = scala.collection.mutable.ArrayBuffer.empty[(String,String)]
    for (term: HanLPTerm <- HanLP.segment(msg)) {
      //println(Array(term.word,term.nature).mkString(":"))
      wordarr += ((term.word,term.nature.name()))
    }
    wordarr.toArray
  }


  def segMsg(msg: String): Array[String] = {
    val wordarr = scala.collection.mutable.ArrayBuffer.empty[String]
    for (term: HanLPTerm <- HanLP.segment(msg) if !term.nature.startsWith('w')) {
      wordarr += term.word
    }
    wordarr.toArray
  }
}
